/**
 * 
 */
package org.metaverse.h2;

import java.io.Closeable;
import java.io.IOException;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Map;
import java.util.Map.Entry;

import javax.sql.DataSource;

import org.metaverse.h2.cache.ValueCache;
import org.metaverse.h2.sql.ValueConstraint2SQL;
import org.metaverse.warehouse.FetchProcedure;
import org.metaverse.warehouse.FetchSpecification;
import org.metaverse.warehouse.Fetcher;
import org.metaverse.warehouse.Senary;
import org.metaverse.warehouse.WarehouseException;
import org.metaverse.warehouse.Senary.Slot;
import org.metaverse.warehouse.com.Constraint;
import org.metaverse.warehouse.com.Equals;
import org.metaverse.warehouse.com.Like;
import org.metaverse.warehouse.com.ValueConstraint;
import org.metaverse.warehouse.com.util.BasicConstraint;
import org.metaverse.warehouse.com.util.ConstraintImpl;
import org.metaverse.warehouse.util.FetchProcedureAdapter;
import org.openrdf.model.ValueFactory;
import org.openrdf.model.impl.ValueFactoryImpl;
import org.singularity.io.CloseableIterator;

/**
 * Fetcher implementation based on H2 DB engine.
 * 
 * @author vjache
 * 
 * @see Fetcher
 * 
 */
public class FetcherImpl implements Fetcher {

	private final DataSource _connPool;
	private final ValueCache _valueCache;

	/**
	 * @param aDataSource
	 * 
	 */
	public FetcherImpl(DataSource aDataSource, ValueCache aNodeCache) {
		_connPool = aDataSource;
		_valueCache = aNodeCache;
	}

	public static void main(String[] args) throws Exception {
		final FetcherImpl fetcher = new FetcherImpl(null, null);
		ValueFactory vf = new ValueFactoryImpl();

		fetcher.fetch(new BasicConstraint().bySubject(
				vf.createURI("http://host#subj-0")).byObject(
				vf.createLiteral("blah")), Slot.Context);

		final ConstraintImpl c = new ConstraintImpl();
		c.setSubject(new Equals(vf.createURI("http://host#subj-0")));
		c.setObject(new Like(vf.createLiteral("blah%")));
		fetcher.fetch(c, Senary.SPOCU);
		
		fetcher.fetch(new FetchProcedureAdapter() {
			
			@Override
			public boolean onNext(Senary aCursor) throws Exception {
				return true;
			}
			
			@Override
			public Constraint getConstraint() {
				// TODO Auto-generated method stub
				return null;
			}
		});
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * org.metaverse.warehouse.Fetcher#fetch(org.metaverse.warehouse.com.Constraint
	 * , org.metaverse.warehouse.Senary.Slot[])
	 */
	@Override
	public CloseableIterator<Senary> fetch(final Constraint aConstraint,
			final Slot... aOrderBy) throws WarehouseException {

		StringBuilder selectSb = procedureToSQL(new FetchProcedureAdapter() {
			@Override
			public Constraint getConstraint() 
			{return aConstraint;}
			@Override
			public Slot[] getOrderBy() 
			{return aOrderBy;}
			@Override
			public Slot[] getSelected() 
			{return null;}
			@Override
			public boolean onNext(Senary aCursor) throws Exception {return true;}
		});
		System.out.println(selectSb);
		final SenaryCacheWarmIterator crd;
		try {
			crd = new SenaryCacheWarmIterator(_connPool, selectSb.toString(),
					_valueCache);
		} catch (SQLException e) {
			throw new WarehouseException(e);
		}
		final SenaryPagedIterator prd = new SenaryPagedIterator(crd);
		return prd;
	}

	private StringBuilder constraintToSQL(Constraint aConstraint,
			boolean aIsCount) {
		final StringBuilder fromSb = new StringBuilder();
		fromSb.append("SENARY ");
		final StringBuilder whereSb = new StringBuilder();
		final Map<Slot, ValueConstraint> all = aConstraint.getAll();
		final ValueConstraint2SQL composer = new ValueConstraint2SQL();
		boolean first = true;
		for (Entry<Slot, ValueConstraint> ent : all.entrySet()) {
			final ValueConstraint constr = ent.getValue();
			if (constr != null) {
				if (first)
					first = false;
				else {
					whereSb.append(" AND ");
				}

				composer.compose(ent.getKey(), constr, whereSb, _valueCache);
				if (composer.isRequireNode()) {
					String alias;
					String fkName;
					switch (ent.getKey()) {
					case Subject:
						alias = "SNODE";
						fkName = "SID";
						break;
					case Predicate:
						alias = "PNODE";
						fkName = "PID";
						break;
					case Object:
						alias = "ONODE";
						fkName = "OID";
						break;
					case Context:
						alias = "CNODE";
						fkName = "CID";
						break;
					case User:
						alias = "UNODE";
						fkName = "UID";
						break;
					default:
						throw new IllegalArgumentException(String.format(
								"Unexpected slot '%s'.", ent.getKey()));
					}

					fromSb.append("JOIN NODE as ").append(alias).append(" ON ")
							.append(fkName).append("=").append(alias).append(
									".NID ");
				}
			}
		}
		StringBuilder selectSb = new StringBuilder();
		if (aIsCount)
			selectSb.append("SELECT COUNT(*) ");
		else
			selectSb.append("SELECT SENARY.* ");
		selectSb.append("FROM ").append(fromSb);
		if (whereSb.length() > 0)
			selectSb.append(" WHERE ").append(whereSb);
		return selectSb;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * org.metaverse.warehouse.Fetcher#count(org.metaverse.warehouse.com.Constraint
	 * )
	 */
	@Override
	public int count(Constraint aConstraint) throws WarehouseException {
		StringBuilder selectSb = constraintToSQL(aConstraint, true);
		System.out.println(selectSb);
		Connection conn = null;
		Statement stmt = null;
		ResultSet rs = null;
		try {
			conn = _connPool.getConnection();
			stmt = conn.createStatement();
			rs = stmt.executeQuery(selectSb.toString());
			rs.next();
			return rs.getInt(1);
		} catch (SQLException e) {
			throw new WarehouseException(e);
		} finally {
			if (rs != null)
				try {
					rs.close();
				} catch (SQLException e) {
					throw new WarehouseException(e);
				}
			if (stmt != null)
				try {
					stmt.close();
				} catch (SQLException e) {
					throw new WarehouseException(e);
				}
			if (conn != null)
				try {
					conn.close();
				} catch (SQLException e) {
					throw new WarehouseException(e);
				}
		}
	}

	@Override
	public ValueFactory getValueFactory() {
		return _valueCache.getValueFactory();
	}

	@SuppressWarnings("unchecked")
	@Override
	public <T extends FetchProcedure> T fetch(FetchProcedure aProcedure)
			throws Exception {
		final StringBuilder selectSb = procedureToSQL(aProcedure);
		System.out.println(selectSb);
		Closeable clz = null;
		try {
			final SenaryCacheWarmIterator crd = new SenaryCacheWarmIterator(
					_connPool, 
					selectSb.toString(), 
					_valueCache,
					aProcedure.getSelected());
			clz = crd;
			final SenaryPagedIterator prd = new SenaryPagedIterator(crd);
			clz = prd;
			while (prd.hasNext()) {
				final boolean proceed = aProcedure.onNext(prd.next());
				if(!proceed)
					break;
			}
			aProcedure.onFinish();
		} catch (SQLException e) {
			throw new WarehouseException(e);
		} finally {
			if (clz != null)
				try {
					clz.close();
				} catch (IOException e) {
					throw new WarehouseException(e);
				}
		}
		return (T) aProcedure;
	}

	private StringBuilder procedureToSQL(FetchSpecification aProcedure) {
		final Slot[] aOrderBy = aProcedure.getOrderBy();
		int limit = aProcedure.getLimit();
		int offset = aProcedure.getOffset();
		final StringBuilder selectSb = constraintToSQL(aProcedure.getConstraint(),
				false);
		if (aOrderBy!=null && aOrderBy.length > 0) {
			StringBuilder orderBySb = new StringBuilder();
			for (int i = 0; i < aOrderBy.length; i++) {
				if (i > 0)
					orderBySb.append(", ");

				switch (aOrderBy[i]) {
				case Subject:
					orderBySb.append("SID");
					break;
				case Predicate:
					orderBySb.append("PID");
					break;
				case Object:
					orderBySb.append("OID");
					break;
				case Context:
					orderBySb.append("CID");
					break;
				case User:
					orderBySb.append("UID");
					break;
				case LastAssertTime:
					orderBySb.append("LAT");
					break;
				default:
					throw new IllegalArgumentException(String.format(
							"Unexpected slot '%s'.", aOrderBy[i]));
				}

			}
			selectSb.append(" ORDER BY ").append(orderBySb);
		}
		
		if(limit>0)
		{
			selectSb.append(" LIMIT ").append(limit);
		}
		else if(offset>0)
		{
			selectSb.append(" LIMIT ").append(Integer.MAX_VALUE);
		}
		
		if(offset>0)
		{
			selectSb.append(" OFFSET ").append(offset);
		}
		return selectSb;
	}

}
